package cn.sciento.transfer.app.service.impl;

import com.alibaba.fastjson.JSON;
import java.util.Collections;
import java.util.Date;
import org.apache.commons.lang3.exception.ExceptionUtils;
import cn.sciento.core.base.BaseConstants;
import cn.sciento.transfer.app.service.ConsumerEventService;
import cn.sciento.transfer.app.service.EventErrorService;
import cn.sciento.transfer.domain.entity.ConsDbConfig;
import cn.sciento.transfer.domain.entity.ConsTenantConfig;
import cn.sciento.transfer.domain.entity.EventError;
import cn.sciento.transfer.domain.repository.EventErrorRepository;
import cn.sciento.transfer.infra.datasource.utils.TransferDataSourceTemplate;
import cn.sciento.transfer.infra.feign.FeignService;
import cn.sciento.transfer.infra.feign.HdttEventErrorService;
import cn.sciento.transfer.infra.rocketmq.handler.MqMessageHandler;
import cn.sciento.transfer.infra.rocketmq.service.MqProducerService;
import cn.sciento.transfer.infra.rocketmq.vo.MqMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

@Service
public class EventErrorServiceImpl implements EventErrorService {
  private static final Logger logger = LoggerFactory.getLogger(EventErrorServiceImpl.class);
  
  @Autowired
  private EventErrorRepository eventErrorRepository;
  
  @Autowired
  private FeignService feignService;
  
  @Autowired
  private MqProducerService mqProducerService;
  
  @Autowired
  private TransferDataSourceTemplate transferDataSourceTemplate;
  
  @Autowired
  private AsyncTaskExecutor asyncTaskExecutor;
  
  private Object syncInsert = new Object();
  
  public void insertEventSendError(EventError eventError) {
    this.eventErrorRepository.insertSelective(eventError);
  }
  
  public EventError processEventError(EventError eventError) {
    ConsDbConfig consDbConfig;
    ConsTenantConfig consTenantConfig;
    ConsumerEventService consumerEventService;
    MqMessageHandler mqMessageHandler;
    eventError.validStatus();
    eventError.setProcessStatus("R");
    this.eventErrorRepository.updateOptional(eventError, new String[] { "processStatus" });
    switch (eventError.getErrorType()) {
      case "SYNC":
        try {
          HdttEventErrorService hdttEventErrorService = (HdttEventErrorService)this.feignService.newInstanceByName(HdttEventErrorService.class, "http://" + eventError.getSourceService());
          Long eventId = hdttEventErrorService.handlerEventError(eventError.getEventId());
          if (eventId != null) {
            updateErrorEventSuccess(eventError, Long.valueOf(eventError.getRetryTimes().longValue() + 1L));
            return eventError;
          } 
        } catch (Exception e) {
          eventError.setProcessMsg(ExceptionUtils.getMessage(e));
        } 
        updateErrorEventFailed(eventError, Long.valueOf(eventError.getRetryTimes().longValue() + 1L));
        break;
      case "SEND":
        try {
          MqMessage mqMessage = (MqMessage)JSON.parseObject(eventError.getMessages(), MqMessage.class);
          int sendOk = this.mqProducerService.sendMsg(mqMessage, "TRANSFER-DEFAULT-PRODUCER");
          if (sendOk == BaseConstants.Flag.YES.intValue()) {
            updateErrorEventSuccess(eventError, Long.valueOf(eventError.getRetryTimes().longValue() + 1L));
            return eventError;
          } 
        } catch (Exception e) {
          eventError.setProcessMsg(ExceptionUtils.getMessage(e));
        } 
        updateErrorEventFailed(eventError, Long.valueOf(eventError.getRetryTimes().longValue() + 1L));
        break;
      case "CONSUME":
        consDbConfig = new ConsDbConfig();
        consTenantConfig = new ConsTenantConfig();
        consTenantConfig.setTenantId(eventError.getSourceTenantId());
        consDbConfig.setConsumerService(eventError.getTargetService());
        consDbConfig.setConsumerTable(eventError.getTargetTable());
        consDbConfig.setConsumerDsId(eventError.getTargetDsId());
        consDbConfig.setConsumerDb(eventError.getTargetDb());
        consDbConfig.setConsumerDs(eventError.getTargetDs());
        consumerEventService = new ConsumerEventServiceImpl(this);
        mqMessageHandler = new MqMessageHandler(consumerEventService, this.transferDataSourceTemplate, consDbConfig, consTenantConfig);
        try {
          mqMessageHandler.initCondition(consDbConfig);
          MqMessage mqMessage = (MqMessage)JSON.parseObject(eventError.getMessages(), MqMessage.class);
          this.transferDataSourceTemplate.doExecute(consDbConfig.getConsumerDs(), consDbConfig.getConsumerService(), jdbcTemplate -> {
                boolean exists = consumerEventService.checkMsgInArch(consDbConfig, mqMessage, jdbcTemplate);
                if (exists)
                  return null; 
                mqMessageHandler.consumerMsg(Collections.singletonList(mqMessage), false, jdbcTemplate);
                return null;
              });
        } catch (Exception e) {
          eventError.setProcessMsg(ExceptionUtils.getMessage(e));
          updateErrorEventFailed(eventError, Long.valueOf(eventError.getRetryTimes().longValue() + 1L));
          return eventError;
        } 
        updateErrorEventSuccess(eventError, Long.valueOf(eventError.getRetryTimes().longValue() + 1L));
        break;
    } 
    return eventError;
  }
  
  public EventError createEventError(EventError eventError) {
    try {
      this.eventErrorRepository.insertSelective(eventError);
    } catch (Exception e) {
      logger.error("eventError EventArch create fail", e);
      return null;
    } 
    return eventError;
  }
  
  public void asyncInsertEventError(EventError eventError) {
    this.asyncTaskExecutor.execute(() -> this.eventErrorRepository.insertSelective(eventError));
  }
  
  private void updateErrorEventSuccess(EventError eventError, Long rt) {
    eventError.setProcessStatus("S");
    eventError.setRetryTimes(rt);
    eventError.setProcessMsg("");
    eventError.setProcessTime(new Date());
    this.eventErrorRepository.updateOptional(eventError, new String[] { "processStatus", "processMsg", "retryTimes", "processTime" });
  }
  
  private void updateErrorEventFailed(EventError eventError, Long rt) {
    eventError.setProcessStatus("E");
    eventError.setRetryTimes(rt);
    eventError.setProcessTime(new Date());
    this.eventErrorRepository.updateOptional(eventError, new String[] { "processStatus", "processMsg", "retryTimes", "processTime" });
  }
}
